package org.example.realtime.jtp.dim.job;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.example.realtime.jtp.common.utils.KafkaUtil;
import org.example.realtime.jtp.dim.function.DimDataFilterFunction;
import org.example.realtime.jtp.dim.function.DimHbaseSinkFunction;
import org.example.realtime.jtp.dim.function.HbaseDimMapFunction;

import java.util.*;

/**
 * @Title: RealtimeSyncDataDimJob
 * @Author Lianzy
 * @Package org.example.realtime.jtp.dim
 * @Date 2025/6/2 19:59
 * @description 按照Flink DataStream流程程序书写方式，其中第3步：获取维表数据和添加字段信息、第4步：将维度数据存储HBase表，使用自定义SinkFunction实现。
 */
public class RealtimeSyncDataDimJob {
    public static void main(String[] args) throws Exception{
        //1.执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //2.数据源
        DataStream<String> stream = KafkaUtil.consumerKafka(env, "topic-db");
        //stream.print();
        //3.数据转换
        DataStream<String> dimStream = handle(stream) ;
//        dimStream.print() ;
        //4.数据接收器
        dimStream.addSink(new DimHbaseSinkFunction());
        //5.触发执行
        env.execute("RealtimeAsyncDataDimJob") ;
    }
    /**
     * 按照表名称过滤获取对应维表数据流，保存HBase表，列簇统一为：info，指定字段名称
     */
    private static DataStream<String> handle(DataStream<String> stream) throws Exception {
        //维度表名称与主键映射
        HashMap<String, String> dimMap = new HashMap<String, String>() {{
            put("base_dic", "dic_code"); put("base_province", "id"); put("base_region", "id");
            put("base_category1", "id"); put("base_category2", "id"); put("base_category3", "id");
            put("sku_info", "id"); put("spu_info", "id"); put("user_info", "id");
            put("activity_info", "id"); put("activity_rule", "id"); put("coupon_info", "id");
            put("base_trademark", "id");
        }};
        //维度表名称
        Set<String> dimSet = new HashSet<>(dimMap.keySet()) ;

        //过滤维表数据
        SingleOutputStreamOperator<String> asyncStream = stream.filter(new DimDataFilterFunction(dimSet));
//        asyncStream.print("async");

        //添加字段值
        SingleOutputStreamOperator<String> hbaseStream = asyncStream.map(new HbaseDimMapFunction(dimMap));
        //hbaseStream.print("hbase");

        //返回结果
        return hbaseStream ;
    }
}
